package rabbitmq

import (
	"context"
	"errors"
	"fmt"
	"github.com/streadway/amqp"
)

// Rabbitmq 初始化rabbitmq连接
type Rabbitmq struct {
	conn *amqp.Connection
	ch   *amqp.Channel // channel对象
	err  error
}

var ErrReject = errors.New("拒绝消息")

// 消费者回调方法
type Consumer func(cxt context.Context, delivery amqp.Delivery)

// New 开始创建一个新的rabitmq连接
func New(config Conf) (*Rabbitmq, error) {
	amqps := fmt.Sprintf("amqp://%s:%s@%s:%d/", config.User, config.Password, config.Host, config.Port)
	conn, err := amqp.Dial(amqps)
	if err != nil {
		return nil, err
	}
	mq := &Rabbitmq{
		conn: conn,
	}

	// 建立channel通道
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	mq.ch = ch

	// 声明exchange交换器
	err = mq.declareExchange("default", nil)
	if err != nil {
		return nil, err
	}
	return mq, nil
}

// declareQueue 定义队列
func (mq *Rabbitmq) declareQueue(name string, args amqp.Table) (q amqp.Queue, err error) {
	q, err = mq.ch.QueueDeclare(
		name,
		true,
		false,
		false,
		false,
		args,
	)

	return q, err
}

// declareQueue 定义交换器
func (mq *Rabbitmq) declareExchange(exchange string, args amqp.Table) error {
	return mq.ch.ExchangeDeclare(
		exchange,
		"direct",
		true,
		false,
		false,
		false,
		args,
	)
}

// PublishQueue 上传消息到queue队列中
func (mq *Rabbitmq) SendMessage(id string, body string) error {
	err := mq.ch.Publish(
		"",    // exchange
		id,    // routing key
		false, // mandatory
		false,
		amqp.Publishing{
			//DeliveryMode: amqp.Persistent,
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	if err != nil {
		return err
	}
	return nil
}

// Start 开始消费消息
func (mq *Rabbitmq) Start(cxt context.Context, id string, fn Consumer, prefetchCount int, consumerCount int) error {
	_, err := mq.declareQueue(id, nil)
	if err != nil {
		return err
	}

	err = mq.ch.Qos(
		prefetchCount, // prefetch count
		0,             // prefetch size
		false,         // global
	)
	if err != nil {
		return err
	}

	for i := 1; i <= consumerCount; i++ {
		go mq.Consume(cxt, id, fn, i)
	}

	return nil
}

func (mq *Rabbitmq) Consume(cxt context.Context, id string, fn Consumer, i int) {
	msgs, err := mq.ch.Consume(
		id,    // queue
		"",    // consumer
		false, // auto-ack
		false, // exclusive
		false, // no-local
		false, // no-wait
		nil,   // args
	)
	if err != nil {
		panic(err)
		return
	}

	for d := range msgs {
		//消费数据
		fn(cxt, d)
		//标记消费
		_ = d.Ack(false)
	}
	return
}

// ConsumeConfirm 从队列中取出数据并且消费(手动确认)
func (mq *Rabbitmq) ConsumeConfirm(cxt context.Context, id string, fn Consumer) error {
	_, err := mq.declareQueue(id, nil)
	if err != nil {
		return err
	}

	err = mq.ch.Qos(
		1,     // prefetch count
		0,     // prefetch size
		false, // global
	)
	if err != nil {
		return err
	}

	msgs, err := mq.ch.Consume(
		id,    // queue
		"",    // consumer
		false, // auto-ack
		false, // exclusive
		false, // no-local
		false, // no-wait
		nil,   // args
	)
	if err != nil {
		return err
	}

	for d := range msgs {
		//消费数据
		fmt.Println(string(d.Body))
		fn(cxt, d)
	}

	return nil
}

func (mq *Rabbitmq) bindQueue(queue, routekey, exchange string) error {
	return mq.ch.QueueBind(
		queue,
		routekey,
		exchange,
		false,
		nil,
	)
}

// Close 关闭链接
func (mq *Rabbitmq) Close() {
	mq.ch.Close()
	mq.conn.Close()
}
